package high.concurrent.consumer;

import com.alibaba.fastjson.JSONObject;
import high.concurrent.esmodel.ProductOrder;
import high.concurrent.esserver.EsServer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Component
public class SpringKafkaListener {

    @Autowired
    private EsServer esServer;
    /**
     * 监听消息
     *  @KafkaListener(topics = {"mysql_table_listener"}, groupId = "listener_mysql")
     *  第一个参数是监听哪些topic，在"springboot-kafka-test"}中列举
     *  第二个参数是给这个监听分配groupId
     *
     * @param record 里面存储接收到的消息的topic, partition, value, key等信息
     * @param acknowledgment  消息手工确认机制
     * @param topic   一般不用
     */
    @KafkaListener(topics = {"order_listener"}, groupId = "listener_mysql")
    public void onMessage(ConsumerRecord<String, String> record, Acknowledgment acknowledgment,
                          @Header(KafkaHeaders.RECEIVED_TOPIC) String topic){

        System.err.println("------------------------------接收到消息（1）---------------------------------------");
        JSONObject jsonObject = JSONObject.parseObject(record.value());
        System.err.println(jsonObject);
        // 进行ES操作
        boolean result = esServer.operateEs(jsonObject, "product_order", ProductOrder.class);
        if(result){
            // 消息确认
            acknowledgment.acknowledge();
        }else{
            acknowledgment.acknowledge();
            // 拒绝消费，比如业务崩了,会把消息扔回队列
            // 重新寻找所有分区，以便在睡眠后重新传递此记录
//            acknowledgment.nack(1);
        }
    }
}
